package com.speciality.demo.demos.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.speciality.demo.demos.dto.FloodDisasterRecVO;
import com.speciality.demo.demos.dto.IrrigationDecisionVO;
import com.speciality.demo.demos.dto.SowingRecommendationVO;
import com.speciality.demo.demos.entity.WeatherHourly;
import com.speciality.demo.demos.mapper.WeatherHourlyMapper;
import com.speciality.demo.demos.service.AgriculturalAnalysisService;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.ml.feature.VectorAssembler;
import org.apache.spark.ml.regression.RandomForestRegressionModel;
import org.apache.spark.ml.regression.RandomForestRegressor;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.expressions.Window;
import org.apache.spark.sql.expressions.WindowSpec;
import org.apache.spark.sql.types.StructType;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.apache.spark.sql.functions.*;
import static org.apache.spark.sql.types.DataTypes.*;

@Slf4j
@Service
public class AgriculturalAnalysisServiceImpl implements AgriculturalAnalysisService {
    //笔记：@Resource是先按照名字注入，之后再按照类型
    @Resource
    private WeatherHourlyMapper weatherHourlyMapper;
    private volatile SparkSession sparkSession;  // volatile保证多线程可见性
    //--------------------------------------------------------getSowingRec需要的参数
    private static final double MIN_TEMP = 10.0;      // 最低日平均温度（℃）
    private static final double MAX_TEMP = 30.0;      // 最高日平均温度（℃）
    private static final double MIN_MOISTURE = 0.6;   // 最低土壤湿度（0~1）
    private static final double MAX_MOISTURE = 0.7;   // 最高土壤湿度（0~1）
    private static final int MAX_RAIN = 10;           // 最大日降水量（mm）
    //---------------------------------------------------------
    private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    /**
     * 初始化Spark会话（线程安全的单例模式）
     */
    private void initializeSpark() {
        if (sparkSession == null) {
            synchronized (this) {  // 双重检查锁保证线程安全
                if (sparkSession == null) {
                    sparkSession = SparkSession.builder()
                            .appName("AgriculturalAnalysis")
                            .master("local[*]")
                            .config("spark.driver.memory", "4g")
                            .config("spark.sql.adaptive.enabled", "true")
                            .getOrCreate();
                    log.info("Spark会话初始化完成");
                }
            }
        }
    }


    /**
     * 灌溉决策核心方法：
     * 1. 分析历史数据（startDate-endDate）
     * 2. 预测endDate之后72小时的土壤湿度变化
     * 3. 生成灌溉建议
     */
    @Override
    public List<IrrigationDecisionVO> getIrrigationDecision(LocalDate startDate, LocalDate endDate) {
        // 1. 初始化Spark会话（单例）
        initializeSpark();

        // 2. 读取历史数据（训练模型用）
        LocalDateTime histStart = LocalDateTime.of(startDate, LocalTime.MIN);
        LocalDateTime histEnd = LocalDateTime.of(endDate, LocalTime.MAX);
        List<WeatherHourly> histList = weatherHourlyMapper.selectByDateRange(startDate, endDate);
        if (histList.isEmpty()) {
            log.warn("历史数据为空，返回空结果");
            return Collections.emptyList();
        }

        // 3. 转换历史数据为Spark Dataset（含Schema定义）
        StructType schema = new StructType()
                .add("time", TimestampType)
                .add("soilMoisture0To7cm", DoubleType)
                .add("et0FaoEvapotranspiration", DoubleType)
                .add("precipitation", DoubleType)
                .add("temperature2m", DoubleType);

        List<Row> dataRows = histList.stream()
                .map(hourly -> RowFactory.create(
                        Timestamp.valueOf(hourly.getTime()),
                        hourly.getSoilMoisture0To7cm(),
                        hourly.getEt0FaoEvapotranspiration(),
                        hourly.getPrecipitation(),
                        hourly.getTemperature2m()
                ))
                .collect(Collectors.toList());
        Dataset<Row> historicalDF = sparkSession.createDataFrame(dataRows, schema);

        //System.out.println("df里面有什么"+historicalDF.toString());

        // 4. 数据清洗与特征工程（含优化的moistureRate计算）
        WindowSpec timeWindow = Window.orderBy("time");
        // 计算填充缺失值的均值（数据完整可省略，保留兼容）
        double avgPrecip = historicalDF.agg(avg("precipitation")).first().getDouble(0);
        double avgEt0 = historicalDF.agg(avg("et0FaoEvapotranspiration")).first().getDouble(0);
        Map<String, Object> fillValues = new HashMap<>();
        fillValues.put("precipitation", avgPrecip);
        fillValues.put("et0FaoEvapotranspiration", avgEt0);

        Dataset<Row> cleanedDF = historicalDF
                .na().fill(fillValues)
                .filter(col("soilMoisture0To7cm").between(0, 1))
                .filter(col("precipitation").geq(0))
                .withColumn("hour", hour(col("time")))
                // 历史数据的moistureRate：用前一小时真实湿度计算（核心）
                .withColumn("moistureRate",
                        (col("soilMoisture0To7cm")
                                .minus(lag("soilMoisture0To7cm", 1).over(timeWindow)))
                                .divide(lit(1)) // 1小时变化率
                )
                .na().fill(Collections.singletonMap("moistureRate", 0.0)); // 首行无历史值，填充0

        // 5. 训练随机森林模型（特征与历史数据一致）
        VectorAssembler assembler = new VectorAssembler()
                .setInputCols(new String[]{
                        "soilMoisture0To7cm",
                        "et0FaoEvapotranspiration",
                        "precipitation",
                        "hour",
                        "moistureRate" // 包含变化率特征
                })
                .setOutputCol("features");

        Dataset<Row> trainingDF = assembler.transform(cleanedDF) // ← 加这行！生成features列
                .withColumn("label", lead("soilMoisture0To7cm", 1).over(timeWindow))
                .na().drop()
                .select("features", "label");


        if (trainingDF.isEmpty()) {
            log.error("训练数据为空，无法生成模型");
            return Collections.emptyList();
        }

        RandomForestRegressor rf = new RandomForestRegressor()
                .setLabelCol("label")
                .setFeaturesCol("features")
                .setNumTrees(10)
                .setMaxDepth(5)
                .setSeed(12345);
        RandomForestRegressionModel model = rf.fit(trainingDF);

        // 6. 读取未来72小时数据（优先用数据库，无则降级）
        LocalDateTime foreStart = histEnd.plusHours(1);
        LocalDateTime foreEnd = foreStart.plusHours(71);
        List<WeatherHourly> foreList = weatherHourlyMapper.selectByDateRange(foreStart.toLocalDate(), foreEnd.toLocalDate());
        log.error(foreList.toString());
        Map<LocalDateTime, WeatherHourly> foreMap = foreList.stream()
                .collect(Collectors.toMap(
                        // 关键修复：将String时间解析为LocalDateTime
                        wh -> LocalDateTime.parse(wh.getTime(), TIME_FORMATTER),
                        Function.identity() // 值保持WeatherHourly对象
                ));

        // 7. 预测循环（核心：动态计算moistureRate）
        List<IrrigationDecisionVO> result = new ArrayList<>(72);
        // 初始化：上一小时湿度 = 历史数据最后一条的湿度
        double prevMoist = histList.get(histList.size() - 1).getSoilMoisture0To7cm();
        double currentMoist = prevMoist; // 当前湿度初始值

        for (int i = 0; i < 72; i++) {
            LocalDateTime forecastTime = foreStart.plusHours(i);
            int hour = forecastTime.getHour();

            // 7.1 获取未来气象数据（优先数据库，无则用固定值）
            WeatherHourly foreData = foreMap.get(forecastTime);
            double precip = foreData != null ? foreData.getPrecipitation() : 0.0;
            double et0 = foreData != null ? foreData.getEt0FaoEvapotranspiration() : 0.5;

            // 7.2 计算当前moistureRate（核心优化：用前一小时湿度 - 当前湿度）
            double moistureRate = (currentMoist - prevMoist) / 1.0; // 1小时变化率

            // 7.3 构造特征向量（含动态变化率）
            Row featureRow = RowFactory.create(
                    currentMoist,    // 当前湿度
                    et0,             // 作物耗水
                    precip,          // 降水
                    hour,            // 小时
                    moistureRate     // 动态变化率（替代固定0）
            );
            Dataset<Row> featureDF = sparkSession.createDataFrame(
                    Collections.singletonList(featureRow),
                    new StructType()
                            .add("soilMoisture0To7cm", DoubleType)
                            .add("et0FaoEvapotranspiration", DoubleType)
                            .add("precipitation", DoubleType)
                            .add("hour", IntegerType)
                            .add("moistureRate", DoubleType)
            );
            Dataset<Row> features = assembler.transform(featureDF).select("features");

            // 7.4 预测下一小时湿度
            double predictedMoist = model.transform(features)
                    .select("prediction")
                    .first()
                    .getDouble(0);

            // 7.5 计算灌溉建议
            double targetMoisture = 0.6;
            double irrigation = predictedMoist < targetMoisture
                    ? Math.round((targetMoisture - predictedMoist) * 1000 * 10) / 10.0
                    : 0.0;
            String riskLevel = irrigation > 50 ? "HIGH"
                    : irrigation > 20 ? "MEDIUM"
                    : "LOW";

            // 7.6 封装结果
            IrrigationDecisionVO vo = new IrrigationDecisionVO();
            vo.setForecastTime(forecastTime);
            vo.setPredictedSoilMoisture(Math.round(predictedMoist * 100) / 100.0);
            vo.setRecommendedIrrigation(irrigation);
            vo.setRiskLevel(riskLevel);
            result.add(vo);

            // 7.7 更新变量：为下一轮准备
            prevMoist = currentMoist;      // 上一小时湿度 = 当前湿度（预测前）
            currentMoist = predictedMoist; // 当前湿度 = 本轮预测值
        }

        return result;
    }

    /**
     * 获取玉米播种建议（按天聚合小时数据）
     * @param startDate 起始日期
     * @param endDate   结束日期
     */
    @Override
    public List<SowingRecommendationVO> getSowingRec(
            LocalDate startDate,
            LocalDate endDate) {
        List<SowingRecommendationVO> result = new ArrayList<>();

        // 遍历时间范围内的每一天
        for (LocalDate date = startDate; date.isBefore(endDate.plusDays(1)); date = date.plusDays(1)) {
            // 1. 查询当天所有小时数据
            List<WeatherHourly> hourlyData = weatherHourlyMapper.selectByDate(date);
            if (hourlyData.isEmpty()) {
                result.add(buildNoDataVO(date));
                continue;
            }

            // 2. 聚合日指标
            double avgTemp = calculateAvgTemp(hourlyData);
            double totalRain = calculateTotalRain(hourlyData);
            double avgMoisture = calculateAvgMoisture(hourlyData);

            // 3. 条件判断
            boolean tempOk = avgTemp >= MIN_TEMP && avgTemp <= MAX_TEMP;
            boolean moistOk = avgMoisture >= MIN_MOISTURE && avgMoisture <= MAX_MOISTURE;
            boolean rainOk = totalRain <= MAX_RAIN;

            // 4. 构建结果
            SowingRecommendationVO vo = new SowingRecommendationVO();
            vo.setDate(date);
            if (tempOk && moistOk && rainOk) {
                vo.setSuitability("适宜");
                vo.setReason("日平均温度、土壤湿度及降水量均符合玉米播种条件。");
                vo.setMeasure("建议当日安排播种，播种后密切关注墒情变化。");
            } else {
                vo.setSuitability("不适宜");
                vo.setReason(buildReason(avgTemp, avgMoisture, totalRain));
                vo.setMeasure(buildMeasure(avgTemp, avgMoisture, totalRain));
            }
            result.add(vo);
        }
        return result;
    }

    // ========== 私有工具方法 ==========
    /**
     * 计算日平均温度（仅统计非空值）
     */
    private double calculateAvgTemp(List<WeatherHourly> hourlyData) {
        List<Double> temps = hourlyData.stream()
                .map(WeatherHourly::getTemperature2m)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        if (temps.isEmpty()) return Double.NaN; // 无有效数据
        return temps.stream().mapToDouble(Double::doubleValue).average().getAsDouble();
    }

    /**
     * 计算日总降水量（null 视为 0）
     */
    private double calculateTotalRain(List<WeatherHourly> hourlyData) {
        return hourlyData.stream()
                .map(h -> h.getPrecipitation() != null ? h.getPrecipitation() : 0.0)
                .mapToDouble(Double::doubleValue)
                .sum();
    }

    /**
     * 计算日平均土壤湿度（仅统计非空值）
     */
    private double calculateAvgMoisture(List<WeatherHourly> hourlyData) {
        List<Double> moistures = hourlyData.stream()
                .map(WeatherHourly::getSoilMoisture0To7cm)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        if (moistures.isEmpty()) return Double.NaN; // 无有效数据
        return moistures.stream().mapToDouble(Double::doubleValue).average().getAsDouble();
    }

    /**
     * 构建无数据的响应
     */
    private SowingRecommendationVO buildNoDataVO(LocalDate date) {
        SowingRecommendationVO vo = new SowingRecommendationVO();
        vo.setDate(date);
        vo.setSuitability("未知");
        vo.setReason("当日无小时级气象数据，无法评估。");
        vo.setMeasure("建议补充数据后重新查询。");
        return vo;
    }

    /**
     * 构建不适宜的原因
     */
    private String buildReason(double avgTemp, double avgMoisture, double totalRain) {
        StringBuilder reason = new StringBuilder();

        // 1. 温度：强制拼接（异常/无异常）
        if (avgTemp < MIN_TEMP || avgTemp > MAX_TEMP) {
            reason.append(String.format(
                    "日平均温度%.1f℃（适宜范围：%s~%s℃）；",
                    avgTemp, MIN_TEMP, MAX_TEMP
            ));
        } else {
            reason.append(String.format(
                    "日平均温度%.1f℃（无异常，适宜范围：%s~%s℃）；",
                    avgTemp, MIN_TEMP, MAX_TEMP
            ));
        }

        // 2. 湿度：强制拼接（异常/无异常）
        if (avgMoisture < MIN_MOISTURE || avgMoisture > MAX_MOISTURE) {
            reason.append(String.format(
                    "土壤湿度%.2f（适宜范围：%s~%s）；",
                    avgMoisture, MIN_MOISTURE, MAX_MOISTURE
            ));
        } else {
            reason.append(String.format(
                    "土壤湿度%.2f（无异常，适宜范围：%s~%s）；",
                    avgMoisture, MIN_MOISTURE, MAX_MOISTURE
            ));
        }

        // 3. 降水：强制拼接（异常/无异常）
        if (totalRain > MAX_RAIN) {
            reason.append(String.format(
                    "日降水量%.1fmm（适宜范围：≤%smm）；",
                    totalRain, MAX_RAIN
            ));
        } else {
            reason.append(String.format(
                    "日降水量%.1fmm（无异常，适宜范围：≤%smm）；",
                    totalRain, MAX_RAIN
            ));
        }

        // 移除最后一个多余的“；”
        if (reason.length() > 0) {
            reason.setLength(reason.length() - 1);
        }

        return reason.toString();
    }

    /**
     * 构建应对措施
     */
    private String buildMeasure(double avgTemp, double avgMoisture, double totalRain) {
        StringBuilder measure = new StringBuilder();
        // 温度问题
        if (avgTemp < MIN_TEMP) {
            measure.append("等待气温回升至10℃以上再播种；");
        } else if (avgTemp > MAX_TEMP) {
            measure.append("避免高温时段播种，或采取遮阳降温措施；");
        }
        // 湿度问题
        if (avgMoisture < MIN_MOISTURE) {
            measure.append("播种前1~2天灌溉保湿（目标湿度≥60%）；");
        } else if (avgMoisture > MAX_MOISTURE) {
            measure.append("提前3~5天排水降湿（目标湿度≤70%）；");
        }
        // 降水问题
        if (totalRain > MAX_RAIN) {
            measure.append("等待降水结束，土壤墒情稳定后再播种；");
        }
        return measure.toString();
    }

    @Override
    public List<FloodDisasterRecVO> getFloodDisasterRec(LocalDate trainStart, LocalDate trainEnd) {
        // 1. 初始化Spark
        initializeSpark();

        // 2. 读取训练数据（一个月的小时级数据）
        List<WeatherHourly> trainHourlyList = getTrainData(trainStart, trainEnd);
        if (trainHourlyList.size() < 24 * 7) { // 至少7天数据才训练
            log.warn("训练数据不足，返回空建议");
            return Collections.emptyList();
        }

        // 3. 数据转换为Spark Dataset，构建特征
        Dataset<Row> trainDF = prepareTrainDataset(trainHourlyList);

        // 4. 训练预测模型（降水量+土壤湿度）
        // 4.1 降水量预测模型
        RandomForestRegressionModel precipModel = trainRegressionModel(
                trainDF,
                "precipitation",  // 预测目标：降水量
                new String[]{"hour", "dayOfWeek", "prev1h_precip", "prev24h_precip_avg"} // 特征
        );
        // 4.2 土壤湿度预测模型
        RandomForestRegressionModel moistModel = trainRegressionModel(
                trainDF,
                "soilMoisture0To7cm",  // 预测目标：土壤湿度
                new String[]{"hour", "dayOfWeek", "precipitation", "prev1h_moist"} // 特征
        );

        // 5. 预测未来三天（72小时）的逐小时数据
        List<WeatherHourly> forecastHourlyList = forecastNext72Hours(
                trainHourlyList,
                precipModel,
                moistModel
        );

        // 6. 聚合为日数据，生成建议
        return generateFloodRec(forecastHourlyList);
    }



    // ========== 私有工具方法 ==========

    /**
     * 读取训练数据（一个月的小时级数据）
     */
    private List<WeatherHourly> getTrainData(LocalDate start, LocalDate end) {
        LambdaQueryWrapper<WeatherHourly> wrapper = new LambdaQueryWrapper<>();
        wrapper.between(WeatherHourly::getTime,
                LocalDateTime.of(start, LocalTime.MIN),
                LocalDateTime.of(end, LocalTime.MAX));
        return weatherHourlyMapper.selectList(wrapper);
    }

    /**
     * 准备训练数据集（添加时间特征、滞后特征）
     */
    private Dataset<Row> prepareTrainDataset(List<WeatherHourly> hourlyList) {
        // 转换为Row（时间、降水量、土壤湿度）
        List<Row> rows = hourlyList.stream()
                .map(h -> RowFactory.create(
                        h.getTime(),
                        h.getPrecipitation() == null ? 0.0 : h.getPrecipitation(),
                        h.getSoilMoisture0To7cm() == null ? 0.0 : h.getSoilMoisture0To7cm()
                ))
                .collect(Collectors.toList());

        // 定义Schema
        // 修改 prepareTrainDataset 方法的 Schema 定义：
        StructType schema = new StructType()
                .add("time", StringType) // 先按字符串读
                .add("precipitation", DoubleType)
                .add("soilMoisture0To7cm", DoubleType);

        Dataset<Row> df = sparkSession.createDataFrame(rows, schema)
                // 显式转换为 Timestamp：假设时间格式是 "yyyy-MM-dd HH:mm:ss"
                .withColumn("time", to_timestamp(col("time"), "yyyy-MM-dd HH:mm:ss"))
                .orderBy("time");

        // 添加时间特征（小时、星期几）
        WindowSpec timeWindow = Window.orderBy("time");
        return df.withColumn("hour", hour(col("time")))
                .withColumn("dayOfWeek", dayofweek(col("time")))
                // 滞后特征：前1小时降水量、前1小时湿度
                .withColumn("prev1h_precip", lag("precipitation", 1).over(timeWindow))
                .withColumn("prev1h_moist", lag("soilMoisture0To7cm", 1).over(timeWindow))
                // 滞后特征：前24小时平均降水量
                .withColumn("prev24h_precip_avg", avg("precipitation").over(
                        Window.orderBy("time").rowsBetween(-23, 0)
                ))
                .na().fill(0.0); // 填充缺失的滞后特征
    }

    /**
     * 训练回归模型（通用方法，可预测降水量或湿度）
     */
    private RandomForestRegressionModel trainRegressionModel(
            Dataset<Row> trainDF,
            String targetCol,  // 目标列名（"precipitation"或"soilMoisture0To7cm"）
            String[] featureCols) {

        // 组装特征向量
        VectorAssembler assembler = new VectorAssembler()
                .setInputCols(featureCols)
                .setOutputCol("features");
        Dataset<Row> featureDF = assembler.transform(trainDF)
                .select("features", targetCol)
                .withColumnRenamed(targetCol, "label");

        // 训练随机森林回归模型
        return new RandomForestRegressor()
                .setLabelCol("label")
                .setFeaturesCol("features")
                .setNumTrees(20)
                .setMaxDepth(5)
                .fit(featureDF);
    }

    /**
     * 预测未来72小时的逐小时数据
     */
    private List<WeatherHourly> forecastNext72Hours(
            List<WeatherHourly> trainData,
            RandomForestRegressionModel precipModel,
            RandomForestRegressionModel moistModel) {

        List<WeatherHourly> forecastList = new ArrayList<>();
        // 最后一条训练数据的时间（作为预测起点）
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        LocalDateTime lastTrainTime = LocalDateTime.parse(
                trainData.get(trainData.size() - 1).getTime(),
                formatter
        );
        // 最后1小时的实际数据（用于初始化滞后特征）
        WeatherHourly lastHour = trainData.get(trainData.size() - 1);
        double lastPrecip = lastHour.getPrecipitation() == null ? 0.0 : lastHour.getPrecipitation();
        double lastMoist = lastHour.getSoilMoisture0To7cm() == null ? 0.0 : lastHour.getSoilMoisture0To7cm();
        // 最后24小时平均降水量（简化：用最后24条训练数据计算）
        double last24hPrecipAvg = calculateLast24hAvg(trainData);

        // 迭代预测72小时
        for (int i = 1; i <= 72; i++) {
            LocalDateTime forecastTime = lastTrainTime.plusHours(i);
            int hour = forecastTime.getHour();
            int dayOfWeek = forecastTime.getDayOfWeek().getValue();

            // 构建当前小时的特征（用前一小时的预测值作为滞后特征）
            Row precipFeature = RowFactory.create(
                    (double)hour,          // int→double
                    (double)dayOfWeek,     // int→double
                    lastPrecip,
                    last24hPrecipAvg
            );
            Row moistFeature = RowFactory.create(
                    (double)hour,          // int→double
                    (double)dayOfWeek,     // int→double
                    lastPrecip,
                    lastMoist
            );

            // 预测当前小时的降水量和湿度
            double predictedPrecip = predictSingleValue(precipModel, precipFeature, new String[]{"hour", "dayOfWeek", "prev1h_precip", "prev24h_precip_avg"});
            double predictedMoist = predictSingleValue(moistModel, moistFeature, new String[]{"hour", "dayOfWeek", "precipitation", "prev1h_moist"});

            // 封装预测结果
            WeatherHourly forecast = new WeatherHourly();
            forecast.setTime(String.valueOf(forecastTime));
            forecast.setPrecipitation(predictedPrecip);
            forecast.setSoilMoisture0To7cm(predictedMoist);
            forecastList.add(forecast);

            // 更新滞后特征（用于下一小时预测）
            last24hPrecipAvg = (last24hPrecipAvg * 24 - lastPrecip + predictedPrecip) / 24; // 滑动平均
            lastPrecip = predictedPrecip;
            lastMoist = predictedMoist;
        }

        return forecastList;
    }

    /**
     * 计算训练数据最后24小时的平均降水量
     */
    private double calculateLast24hAvg(List<WeatherHourly> trainData) {
        int startIdx = Math.max(0, trainData.size() - 24);
        return trainData.subList(startIdx, trainData.size()).stream()
                .mapToDouble(h -> h.getPrecipitation() == null ? 0.0 : h.getPrecipitation())
                .average()
                .orElse(0.0);
    }

    /**
     * 单条数据预测（辅助方法）
     */
    private double predictSingleValue(
            RandomForestRegressionModel model,
            Row featureRow,
            String[] featureCols) {

        StructType schema = new StructType();
        for (String col : featureCols) {
            schema = schema.add(col, DoubleType);
        }

        Dataset<Row> featureDF = sparkSession.createDataFrame(Collections.singletonList(featureRow), schema);
        VectorAssembler assembler = new VectorAssembler()
                .setInputCols(featureCols)
                .setOutputCol("features");
        return model.transform(assembler.transform(featureDF))
                .select("prediction")
                .first()
                .getDouble(0);
    }

    /**
     * 生成涝灾建议（复用FloodDisasterRecVO）
     */
    private List<FloodDisasterRecVO> generateFloodRec(List<WeatherHourly> forecastHourlyList) {
        // 按日期分组聚合
        Map<LocalDate, List<WeatherHourly>> dailyMap = forecastHourlyList.stream()
                .collect(Collectors.groupingBy(h -> {
                    // 假设time格式："yyyy-MM-dd HH:mm:ss"，截取日期部分解析
                    String dateStr = h.getTime().substring(0, 10); // 取"2025-10-31"
                    return LocalDate.parse(dateStr);
                }));

        List<FloodDisasterRecVO> result = new ArrayList<>();
        for (Map.Entry<LocalDate, List<WeatherHourly>> entry : dailyMap.entrySet()) {
            LocalDate date = entry.getKey();
            List<WeatherHourly> dailyData = entry.getValue();

            // 日总降水量（mm）
            double totalPrecip = dailyData.stream()
                    .mapToDouble(WeatherHourly::getPrecipitation)
                    .sum();
            // 日平均土壤湿度
            double avgMoist = dailyData.stream()
                    .mapToDouble(WeatherHourly::getSoilMoisture0To7cm)
                    .average()
                    .orElse(0.0);

            // 生成建议（阈值可调整）
            FloodDisasterRecVO vo = new FloodDisasterRecVO();
            vo.setDate(date);
            if (totalPrecip > 50 || avgMoist > 0.75) {
                vo.setRiskLevel("高风险");
                vo.setRiskDesc(String.format("预测日总降水%.1fmm，平均土壤湿度%.2f，易发生涝灾", totalPrecip, avgMoist));
                vo.setSuggestion("启动紧急排水，疏通田间沟渠，转移低洼处作物");
            } else if (totalPrecip > 30 || avgMoist > 0.65) {
                vo.setRiskLevel("中风险");
                vo.setRiskDesc(String.format("预测日总降水%.1fmm，平均土壤湿度%.2f，存在涝灾隐患", totalPrecip, avgMoist));
                vo.setSuggestion("检查排水设施，准备应急设备，减少灌溉");
            } else {
                vo.setRiskLevel("低风险");
                vo.setRiskDesc(String.format("预测日总降水%.1fmm，平均土壤湿度%.2f，涝灾风险低", totalPrecip, avgMoist));
                vo.setSuggestion("正常田间管理，持续监测气象变化");
            }
            result.add(vo);
        }

        // 按日期排序
        return result.stream()
                .sorted(Comparator.comparing(FloodDisasterRecVO::getDate))
                .collect(Collectors.toList());
    }


}
